package com.goodwill.androidtest.rxjava;


import com.goodwill.androidtest.Person;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.BackpressureStrategy;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.FlowableEmitter;
import io.reactivex.rxjava3.core.FlowableOnSubscribe;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.core.Single;
import io.reactivex.rxjava3.core.SingleObserver;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class RxJavaTest {
    public static void main(String[] args) {
//        justObservable();
//        fromArrayObservable();
//        justSingle();
//        createFlowable();
    }

    private static synchronized void createFlowable() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<Integer> emitter) throws Throwable {
                for (int i = 0; ; i++) {
                    emitter.onNext(i);
                }
            }
        }, BackpressureStrategy.BUFFER).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Throwable {
                System.out.println("Flowable.create --> accept:" + integer.intValue());
            }
        });


    }

    private static void justSingle() {
        Single.just("1")
                .subscribe(new SingleObserver<String>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {

                    }

                    @Override
                    public void onSuccess(@NonNull String s) {
                        System.out.println("Single.just -->onSuccess: " + s);
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }
                });
    }

    private static void fromArrayObservable() {
        Observable.fromArray(new Person("张三", 12), new Person("李四", 23), new Person("王五", 45))
//                .subscribeOn(Schedulers.io())
                .map(new Function<Person, Integer>() {
                    @Override
                    public Integer apply(Person person) throws Throwable {
                        System.out.println("function apply:" + person.toString());
                        return person.getAge();
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Throwable {
                System.out.println("accept:" + integer);
            }
        });
    }

    private static void justObservable() {
        Observable.just("guo", "xiao", "jun")
                .subscribeOn(Schedulers.io())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        System.out.println("just----onSubscribe:" + d.isDisposed());
                    }

                    @Override
                    public void onNext(@NonNull String s) {
                        System.out.println("just----onNext:" + s);
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {
                        System.out.println("just----onError:" + e.getMessage());
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("just----onComplete:");
                    }
                });
    }
}
